package com.newland.nxsms.cmpp;

import java.util.Iterator;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.newland.nxsms.NxSmsMain;
import com.newland.nxsms.kafka.KafkaConsumer;
import com.newland.nxsms.utils.StringUtils;



public class CollectData implements Runnable {


	public static final Logger log = LoggerFactory.getLogger(CollectData.class);
	private KafkaConsumer consumer;
	private String bootstrap;
	private String topic;
	private String groupId;
	public CollectData(String bootstrap,String topic,String groupId) {
		this.bootstrap = bootstrap;
		this.topic = topic;
		this.groupId = groupId;
	}

	private void init() {
		consumer = new KafkaConsumer(bootstrap,groupId);
		consumer.setBatchDurationMillis(100L);
		List<String> topics = StringUtils.split(topic, ",");
		consumer.setTopic(topics);
		consumer.connect();
	}
	@Override
	public void run() {
		init();
		log.info("start get data from kafka");
		while(true) {
			Iterator<ConsumerRecord<String, byte[]>> it = consumer.poll();
			while(it.hasNext()) {
				try{

					String msg = new String(it.next().value());
					log.info("-----data-----"+msg);
					NxSmsMain.dealData.add(msg);
				} catch (Exception e) {
					log.error("get msg error!",e);
				}
			}
		}
	}

	private void close() {
		if (null != consumer) {
			consumer.close();
		}
	}
}


